package socket.socket;


import io.reactivex.schedulers.Schedulers;
import socket.modle.ConnectedCallback;
import socket.modle.PlcInterface;
import socket.util.ByteUtil;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.ArrayList;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;


/**
 * Created by cjx on 2019/8/25
 * 说明：次方案只针对PLC做客户端并且是1对1的socket方案，
 */
public class SocketClientUtil implements PlcInterface {


    private ConnectedCallback callback; // socket连接状态上报
    private Socket clientSocket; // 连接进来的socket客户端， 当前只有最多1个客户端
    private DataOutputStream dataOutputStream; // socket的输出流
    private DataInputStream dataInputStream; // socket的输入流

    private int retryCount = 0; // socket服务器创建失败重试次数
    private int currentRetryCount = 0; // 当前重试的次数
    private Object[] mparams;

    ConcurrentLinkedQueue<byte[]> sendDataQueue = new ConcurrentLinkedQueue<>();

    public SocketClientUtil() {

    }

    @Override
    public PlcInterface init(ConnectedCallback callback, int retryCount) {
        this.callback = callback;
        this.retryCount = retryCount;
        return this;
    }

    @Override
    public void connect(final Object... params) {
        if (isConnected()) {
            return;
        }

        Schedulers.computation().scheduleDirect(new Runnable() {
            @Override
            public void run() {
                final String host = (String) params[0];
                final int port = (int) params[1];
                try {
                    clientSocket = new Socket(host, port);
                    try {
                        dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                        dataInputStream = new DataInputStream(clientSocket.getInputStream());
                        Thread thread = new Thread(new ReceiverRunnable(clientSocket));
                        thread.setPriority(10);
                        thread.start();
                        Thread sendT = new Thread(new SendRunable());
                        sendT.setPriority(9);
                        sendT.start();
                        callback.onConnectedStatusChange("", true, null, "连接成功");
                    } catch (IOException e) {
                        e.printStackTrace();
                        callback.onConnectedStatusChange("", false, e, "创建socket失败");
                        retryConnect(params);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                    callback.onConnectedStatusChange("", false, e, "创建socket失败");
                    retryConnect(params);
                }
            }
        });
    }

    @Override
    public void sendData(String key, final byte[] data) {
        if (isConnected()) {
            sendDataQueue.add(data);
           // System.out.println("sendDataQueue  产生后:"+sendDataQueue.size());
        }
    }

    @Override
    public boolean isConnected() {
        return clientSocket != null && clientSocket.isConnected() && !clientSocket.isClosed();
    }

    @Override
    public void disConnect(Exception e) {
        if (clientSocket == null) {
            return;
        }
        try {
            dataInputStream.close();
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        try {
            dataOutputStream.close();
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        try {
            clientSocket.close();
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        clientSocket = null;
        sendDataQueue.clear();
        callback.onConnectedStatusChange("", false, e, "断开socket连接");
    }

    @Override
    public void destroy(String who) {
        disConnect(null);
    }


    private Runnable reConnectRun = new Runnable() {
        @Override
        public void run() {
            connect(mparams);
        }
    };

    private void retryConnect(Object... params) {
        if (retryCount < 1 || currentRetryCount >= retryCount) {
            return;
        }
        currentRetryCount++;
        mparams = params;
        Schedulers.io().scheduleDirect(reConnectRun, 2, TimeUnit.SECONDS);
    }


    private class SendRunable implements Runnable {

        @Override
        public void run() {
            while (isConnected()) {
                try {
                    if (sendDataQueue.size() > 0) {
                        int readSize = Math.min(15, sendDataQueue.size());
                        ArrayList<byte[]> getbytes = new ArrayList<>(readSize);
                        int len = 0;
                        for(int i = 0; i<readSize;i++){
                           byte[] bs = sendDataQueue.poll();
                            len += bs.length;
                            getbytes.add(bs);
                        }
                        byte[] data = new byte[len];
                        int index = 0;
                     //   StringBuilder stringBuilder = new StringBuilder();
                        for(byte[] s: getbytes){
                            System.arraycopy(s, 0, data, index, s.length);
                            index+=s.length;
//                            for(byte b: s){
//                                stringBuilder.append(ByteUtil.byteToHexString(b)+" ");
//                            }
//                            int l = ByteUtil.getInt(new byte[]{s[3], s[4], s[5], s[6]});
//                            if(l + 7!=s.length){
//                                System.out.println("发送有诡异 ");
//                            }
//                            stringBuilder.append("\n");
                        }
                        try {
                            dataOutputStream.write(data);
                            dataOutputStream.flush();
                          //  System.out.println(clientSocket.getLocalPort()+" :"+System.currentTimeMillis()+"[TIME] 发送 "+stringBuilder.toString());
                        } catch (IOException e) {
                            e.printStackTrace();
                            destroy("");
                        }
                        //System.out.println(System.currentTimeMillis()+"sendDataQueue  消费后:"+sendDataQueue.size());
                    }
                    Thread.sleep(100);
                } catch (Exception e) {
                    e.printStackTrace();
                    destroy("");
                }
            }
        }
    }

    /**
     * 接受socket发送来的消息的运行类
     */
    private class ReceiverRunnable implements Runnable {

        Socket socket;

        ReceiverRunnable(Socket socket) {
            this.socket = socket;
        }

        @Override
        public void run() {
            while (socket.isConnected() && !socket.isClosed()) {
                try {
                    callback.onReceive("", dataInputStream);
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            disConnect(null);
        }
    }
}
